package com.cloud.kafka;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class BatchMessageListener {

    @KafkaListener(containerFactory = "kafkaListenerContainerFactory", topics = "batchTest", groupId = "your-group")
    public void listen(List<String> messages, Acknowledgment acknowledgment) {
        System.out.println("Received messages: " + messages);
        // 处理消息集合
        try {
            for (String message : messages) {
                System.out.println("Processing message: " + message);
                // 模拟消息处理逻辑
                // 如果消息处理成功，则手动确认
                acknowledgment.acknowledge();
            }
        } catch (Exception e) {
            // 如果消息处理失败，可以选择不调用 acknowledgment.acknowledge()，让 Kafka 重新投递消息
            System.out.println("Message processing failed. Will retry.");
        }
    }
}
